package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/11/18
 *
 *  数据源必须是无界流： 有始无终。 从某一刻开始，源源不断地产生数据！
 *
 *  通过网络端口模拟无限的数据。
 *          安装netcat :  sudo yum -y install nc
 *
 *          启动nc服务端:  nc -lk 主机名 端口号
 *
 *          无界流和有界流的区分在于是否是源源不断地产生数据和计算数据！
 *                  有界流，由于数据是有限的，计算完成就退出！
 *                  无界流，有始无终，数据源源不断，计算一直运行，直到数据源无法连接！
 */
public class Demo3_UnBoundedStream
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置并行度为1，全局只有一个Task计算
        env.setParallelism(1);
        //读取数据
        DataStreamSource<String> ds = env.socketTextStream("hadoop103", 8888);

        //转换
        ds.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>()
        {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word,1));
                }
            }
        })
          //keyBy： 指定一个字段，按照这个字段作为key进行分组
          .keyBy(new KeySelector<Tuple2<String, Integer>, String>()
          {
              //返回的数据作为key，会被分组
              @Override
              public String getKey(Tuple2<String, Integer> value) throws Exception {
                  //f0代表tuple2的第一个字段
                  return value.f0;
              }
          })
          .sum(1)
          .print();


        //启动执行环境
        env.execute();

    }
}
